package com.linkstec.mot.main;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.linkstec.mot.service.KafkaConsumerService;
import com.linkstec.mot.util.KafkaConsumerToRedis;

public class syncApplicationStarter {

	public static void main(String[] args) {

		System.out.println("start KafkaConsumerToRedis...");
		int numConsumers = 1;
		String groupId = "pushTest_2";
		List<String> topics = Arrays.asList("T_GP", "HUAXI_LINGZHI_SMOT_POC");
		ExecutorService executor = Executors.newFixedThreadPool(numConsumers);
		final List<KafkaConsumerToRedis> consumers = new ArrayList<>();
		for (int i = 0; i < numConsumers; i++) {
			KafkaConsumerToRedis consumer = new KafkaConsumerToRedis(i, groupId, topics);
			consumers.add(consumer);
			executor.submit(consumer);
		}
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				for (KafkaConsumerToRedis consumer : consumers) {
					consumer.shutdown();
				}
				executor.shutdown();
				try {
					executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		});
		// root
		// application
		System.out.println("start kafkaConsumerService...");
		ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext-common.xml");
		KafkaConsumerService kafkaConsumerService = (KafkaConsumerService) context.getBean("kafkaConsumerService");
		kafkaConsumerService.start();

		System.out.println("启动结束");
	}
}